package com.xinqi.modules.mq.delay;

import com.xinqi.common.base.utils.date.Dates;
import com.xinqi.common.rocketmq.annotation.MqConsumer;
import com.xinqi.common.rocketmq.dto.MqDelayMessage;
import com.xinqi.common.rocketmq.util.ByteUtils;
import com.xinqi.common.rocketmq.util.MqHelpUtils;
import com.xinqi.modules.mq.delay.dao.MqDelayMessageRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
import org.apache.rocketmq.client.apis.message.MessageBuilder;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Date;
import java.util.UUID;

/**
 * 延时队列消费者
 *
 * @author linkai
 * @date 2023/11/19
 */
@Slf4j
@Component
@RequiredArgsConstructor
@MqConsumer(topic = "${rocketmq.delayTopic}", groupName = "exam_delay")
public class MqDelayMessageConsumer implements MessageListener {

    @Value("${rocketmq.delayTopic}")
    private String delayTopic;
    /**
     * 延时消息，最大支撑24小时内消息
     */
    private static final Long MAX_DELAY_MILLI = Duration.ofHours(24).toMillis();

    private final MqDelayMessageRepository mqDelayMessageRepository;

    @Qualifier("defaultProducer")
    private final Producer producer;

    public void sendDelayMessage(MqDelayMessage message) {
        message.setId(UUID.randomUUID().toString());
        // 添加数据库
        mqDelayMessageRepository.insert(message.getId(), message.getKey(),
            Dates.long2LocalDateTime(message.getArrivalTimeMilli()));
        sendMessage(message);
    }

    /**
     * 接收延时任务消息
     */
    @Override
    public ConsumeResult consume(MessageView messageView) {
        MqDelayMessage message = ByteUtils.byteToObj(messageView.getBody(), MqDelayMessage.class);
        message.addCycleTimes();
        sendMessage(message);
        return ConsumeResult.SUCCESS;
    }

    private void sendMessage(MqDelayMessage message) {
        // 校验消息是否被删除
        if (mqDelayMessageRepository.checkDel(message.getId())) {
            mqDelayMessageRepository.updateStateByMqId(message.getId());
            return;
        }
        long delayTimeMill = getDelayTimeMill(message.getArrivalTimeMilli());
        log.info("====> 消费-延时消息发送，messageId = {}, cycleTimes = {}, key = {}, targetTopic = {}, arrivalTime = {}, delayTimeMill = {}",
            message.getId(), message.getCycleTimes(), message.getKey(), message.getTopic(),
            DateFormatUtils.format(new Date(message.getArrivalTimeMilli()), "yyyy-MM-dd HH:mm:ss"),
            delayTimeMill);
        ClientServiceProvider provider = MqHelpUtils.provider();
        MessageBuilder messageBuilder = provider.newMessageBuilder();
        if (delayTimeMill > 0) {
            messageBuilder
                .setTopic(delayTopic)
                .setKeys(message.getId())
                .setTag(message.getTag())
                .setDeliveryTimestamp(System.currentTimeMillis() + delayTimeMill)
                .setBody(ByteUtils.obj2Byte(message));
        } else {
            messageBuilder
                .setTopic(message.getTopic())
                .setKeys(message.getKey())
                .setTag(message.getTag())
                .setBody(ByteUtils.str2Byte(message.getBody()));
            log.info("====> 到期消息发送，topic = {}, key = {}, tag = {}, body = {}",
                message.getTopic(), message.getKey(), message.getTag(), message.getBody());
            mqDelayMessageRepository.updateStateByMqId(message.getId());
        }
        try {
            producer.send(messageBuilder.build());
        } catch (ClientException e) {
            log.error("消费-延时消息失效，messageId = {}, cycleTimes = {}, key = {}, targetTopic = {}, arrivalTimeMilli = {}, delayTimeMill = {}",
                message.getId(), message.getCycleTimes(), message.getKey(), message.getTopic(), message.getArrivalTimeMilli(), delayTimeMill);
            throw new RuntimeException(e);
        }
    }

    private long getDelayTimeMill(Long arrivalTimeMilli) {
        long l = arrivalTimeMilli - System.currentTimeMillis();
        // 一秒内的都立即发送
        if (l < 1000) {
            return 0;
        }
        if (l > MAX_DELAY_MILLI) {
            return MAX_DELAY_MILLI;
        }
        return l;
    }

}
